package cn.doit.common

import org.apache.spark.sql.Row


object DolphinKPI {
  /**
    * 整理判断业务指标数据
    */

  def caculateKpi(row: Row): (List[Double]) = {
    //获取省市及日期
    val day = row.getAs[String]("requestdate").substring(0, 10)
    val province = row.getAs[String]("provincename")
    val city = row.getAs[String]("cityname")
    //获取判断要用的字段
    val requestmode = row.getAs[Int]("requestmode")
    val processnode = row.getAs[Int]("processnode")
    val iseffective = row.getAs[Int]("iseffective")
    val isbilling = row.getAs[Int]("isbilling")
    val isbid = row.getAs[Int]("isbid")
    val iswin = row.getAs[Int]("iswin")
    val adorderid = row.getAs[Int]("adorderid")
    val winprice = row.getAs[Double]("winprice") / 1000d
    val adpayment = row.getAs[Double]("adpayment") / 1000d


    //对字段进行分组,累加等操作
    val rawReq   = if (requestmode == 1 && processnode >= 1) 1 else 0
    val effReq   = if (requestmode == 1 && processnode >= 2) 1 else 0
    val adReq    = if (requestmode == 1 && processnode >= 3) 1 else 0
    val isRTB    = if (iseffective == 1 && isbilling == 1 && isbid == 1 && adorderid != 0) 1 else 0
    val isuccRTB = if (iseffective == 1 && isbilling == 1 && iswin == 1) 1 else 0
    val adShow   = if (requestmode == 2 && iseffective == 1) 1 else 0
    val adClik   = if (requestmode == 3 && iseffective == 1) 1 else 0
    val expense  = if (iseffective == 1 && isbilling == 1 && iswin == 1) winprice  else 0d
    val cost     = if (iseffective == 1 && isbilling == 1 && iswin == 1) adpayment    else 0d

     List(rawReq, effReq, adReq.toDouble, isRTB, isuccRTB, adShow, adClik, expense, cost)

  }
}
